package com.atguigu.actual.edu0417.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.beans.TableProcess;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.unils.DimInfoUtil;
import com.atguigu.actual.edu0417.unils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author: techyang
 * @since: 2023-10-19 20:34
 * @description: DimSink
 **/
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    // HBase连接
    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(conn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {

        // 数据操作类型 insert,update,delete
        String type = jsonObj.getString("type");
        // 配置信息
        TableProcess tableProcess = jsonObj.getObject("tableProcess", TableProcess.class);
        // HBase中对应的表名
        String sinkTable = tableProcess.getSinkTable();
        String sinkRowKey = tableProcess.getSinkRowKey();
        String rowKey = jsonObj.getString(sinkRowKey);
        if ("delete".equals(type) || ("update".equals(type) && "1".equals(jsonObj.getString("deleted")))) {
            // 物理删除 或者 逻辑删除 删除hbase中对应的数据  ⚠️⚠️⚠️ rowkey 不要直接取配置表的值,要根据配置的字段取jsonObj里边的值
            HbaseUtil.delRow(conn, EduConfig.HBASE_NAMESPACE,sinkTable,rowKey);
        } else {
            // 插入新数据 或者 更新数据  往hbase中put数据
            // 列族
            String[] sinkFamilies = tableProcess.getSinkFamily().split(",");
            // 列
            String[] sinkColumns = tableProcess.getSinkColumns().split(",");
            // 列对应的值
            String[] sinkValues = new String[sinkColumns.length];
            for (int i = 0; i < sinkColumns.length; i++) {
                sinkValues[i] = jsonObj.getString(sinkColumns[i]);
            }
            // ⚠️⚠️⚠️ rowkey 不要直接取配置表的值,要根据配置的字段取jsonObj里边的值
            HbaseUtil.putRow(conn,EduConfig.HBASE_NAMESPACE,sinkTable,rowKey,sinkFamilies[0],sinkColumns,sinkValues);
        }

        if ("delete".equals(type) || "update".equals(type)) {
            DimInfoUtil.delRedisKey(sinkTable + ":" + rowKey);
        }
    }
}
